package com.sky.rocket.consumer.listener;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;


/**
 * 这里是所有消息统一监听
 * 这种监听配合config一起使用
 */
@Component
public class RocketMessageListener implements MessageListenerConcurrently {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        if(CollectionUtils.isEmpty(list)){
            System.out.println("接收到的消息为空，不做任何处理");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt messageExt = list.get(0);
        String msg = new String(messageExt.getBody());
        System.out.println("接收到的消息是："+msg);
        if(messageExt.getTopic().equals("rocketKey")){
            if(messageExt.getTags().equals("name")){
                System.out.println("这里是topic=rocketKey, tag=name");
                int reconsumeTimes = messageExt.getReconsumeTimes();
                // 消息已经重试了3次，如果不需要再次消费，则返回成功
                if(reconsumeTimes == 3){
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                // 处理对应的业务逻辑
            }
            if (messageExt.getTags().equals("sky")) {
                System.out.println("这里是topic=rocketKey, tag=sky");
            }
        } else if (messageExt.getTopic().equals("skyTopic")) {
            if (messageExt.getTags().equals("name")) {
                System.out.println("这里是topic=sky, tag=name");
            }
        }

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
